package com.bd08.flink.demo

import org.apache.flink.streaming.api.scala.{StreamExecutionEnvironment, createTypeInformation}

object StateDemo  extends App {
  val env = StreamExecutionEnvironment.getExecutionEnvironment
  val ds=env.fromElements((1,3),(1,5),(1,7),(1,4),(1,2))
  val counts=ds.keyBy(_._1).mapWithState((in:(Int,Int),count:Option[Int])=>
    count match{
      case Some(c)=>((in._1,c),Some(c+in._2))
      case None=>((in._1,0),Some(in._2))
    })
  counts.print
  env.execute


}
